package com.xueyuan.wata.daph.api.node

import com.xueyuan.wata.daph.api.GlobalContext
import com.xueyuan.wata.daph.api.config.{NodeConfig, NullNodeConfig}
import org.apache.logging.log4j.scala.Logging
import org.json4s.jackson.JsonMethods

abstract class Node extends java.io.Serializable with Logging {
  private var _nodeDescription: NodeDescription = _
  private var _config: NodeConfig = _
  private var _gc: GlobalContext = _

  def init(nodeDescription: NodeDescription, gc: GlobalContext): java.util.List[NodeInitErrorResult] = {
    val res = new java.util.ArrayList[NodeInitErrorResult]()
    _nodeDescription = nodeDescription
    try {
      _config = {
        if (getNodeConfigClass != null) {
          JsonMethods.mapper.convertValue(
            nodeDescription.config,
            getNodeConfigClass
          )
        } else {
          new NullNodeConfig()
        }
      }
    } catch {
      case e: Throwable =>
        res.add(NodeInitErrorResult(
          this.getClass.getSimpleName,
          "NodeConfig",
          s"${getNodeConfigClass.getName}",
          s"NodeConfig实例化失败！${e.getMessage}"
        ))
    }
    _gc = gc

    res
  }

  final def runNode(): Unit = {
    try {
      run()
    } catch {
      case e: Throwable =>
        logger.error(s"Node run failed! ${e.getMessage}")
        throw e
    }
  }

  protected def run(): Unit

  def viewNode(): Unit = {}

  def getNodeConfigClass: Class[_ <: NodeConfig] = classOf[NullNodeConfig]

  final def id: String = nodeDescription.id

  final def nodeDescription: NodeDescription = _nodeDescription

  final def nodeConfig: NodeConfig = _config

  final def gc: GlobalContext = _gc

  final protected def setDSToLine(ds: Any, line: String): Unit = {
    gc.storage.setDSToLine(ds, line)
  }

  final protected def setDSToLines(ds: Any, lines: Array[String]): Unit = {
    gc.storage.setDSToLines(ds, lines)
  }

  final protected def getDSByLine(line: String): Any = {
    gc.storage.getDSByLine(line)
  }

  final protected def getLineToDS: Map[String, Any] = {
    gc.storage.getLineToDSByLines(nodeDescription.inLines)
  }
}
